package com.spark.WordCount;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.*;
import scala.Tuple2;

import java.util.*;

public class TopNByJava {
    public static void main(String[] args) {
        SparkConf sparkConf=new SparkConf()
                .setMaster("local")
                .setAppName("TopNByJava");
        JavaSparkContext sparkContext=new JavaSparkContext(sparkConf);

        //1、获取主播的开播记录和用户送礼记录--使用fastjson包解析
        //主播开播记录(video_info.log):主播ID：uid，直播间ID：vid，大区：area  --- (vid,(uid,area))
        //用户送礼记录(gift_record.log)：直播间ID：vid，金币数量：gold ---(vid,gold)
        JavaRDD<String> VideoInfoRdd=sparkContext.textFile("datas/video_info.log");
        JavaRDD<String> GiftRecordRdd=sparkContext.textFile("datas/gift_record.log");
        JavaPairRDD<String, Tuple2<String, String>> videoPairRdd=VideoInfoRdd.mapToPair(new PairFunction<String, String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, Tuple2<String, String>> call(String s) throws Exception {
                JSONObject jsonObject= JSON.parseObject(s);
                String vid=jsonObject.getString("vid");
                String uid=jsonObject.getString("uid");
                String area=jsonObject.getString("area");
                return new Tuple2<>(vid,new Tuple2<>(uid,area));
            }
        });
        JavaPairRDD<String ,Integer> giftPairRdd=GiftRecordRdd.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                JSONObject jsonObject=JSON.parseObject(s);
                String vid=jsonObject.getString("vid");
                Integer gold=Integer.parseInt(jsonObject.getString("gold"));
                return new Tuple2<>(vid,gold);
            }
        });
        //2、对用户送礼记录数据进行聚合，对相同vid的数据求和，
        // 计算主播所在直播间的音浪收入（用户可能在一次直播中给主播送多次礼物） --（vid,gold_sum）,
        JavaPairRDD<String,Integer> giftPairByReduceRdd= giftPairRdd.reduceByKey(new Function2<Integer,Integer, Integer>() {
            @Override
            public Integer call(Integer i1, Integer i2) throws Exception {
                return i1+i2;
            }
        });
        //3、把这两份数据join到一块，vid作为join的key--(vid,((uid,area),gold_sum))
        JavaPairRDD<String, Tuple2<Tuple2<String, String>, Integer>> joinRdd=videoPairRdd.join(giftPairByReduceRdd);
        //4、使用map迭代join之后的数据，最后获取到uid、area、gold_sum字段-- ((uid,area),gold_sum)
        JavaPairRDD JoinMapRdd=joinRdd.mapToPair(new PairFunction<Tuple2<String, Tuple2<Tuple2<String, String>, Integer>>, Tuple2<String,String >, Integer>() {
            @Override
            public Tuple2<Tuple2<String, String>, Integer> call(Tuple2<String, Tuple2<Tuple2<String, String>, Integer>> stringTuple2) throws Exception {
                String uid=stringTuple2._2._1._1;
                String area=stringTuple2._2._1._2;
                Integer gold_sum=stringTuple2._2._2;
                return new Tuple2<>(new Tuple2<>(uid,area),gold_sum);
            }
        });
        //5、使用reduceByKey算子对数据进行聚合（可能存在地区的主播开播多次有多个直播间的音浪收入）-- ((uid,area),gold_sum_all)
        JavaPairRDD reduceByKeyRdd=JoinMapRdd.reduceByKey(new Function2<Integer,Integer,Integer>() {
            @Override
            public Integer call(Integer o, Integer o2) throws Exception {
                return o+o2;
            }
        });
        //6、接下来对需要使用groupByKey对数据进行分组，所以先使用map进行转换
        // map：(area,(uid,gold_sum_all))
        // groupByKey: area,<(uid,gold_sum_all),(uid,gold_sum_all),(uid,gold_sum_all)
        JavaPairRDD groupRdd=reduceByKeyRdd.mapToPair(new PairFunction<Tuple2<Tuple2<String,String>,Integer>,String,Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Tuple2<String, Integer>> call(Tuple2<Tuple2<String, String>, Integer> tuple2Tuple2) throws Exception {
                String uid=tuple2Tuple2._1._1;
                String area=tuple2Tuple2._1._2;
                Integer gold_Sum_all=tuple2Tuple2._2;
                return new Tuple2<>(area,new Tuple2<>(uid,gold_Sum_all));
            }
        }).groupByKey();//分组之后的数据是一个Iterable迭代器
        //7、使用map迭代每个分组内的数据，按照金币数量倒序排序，取前N个，最终输出area,topN
        //这个topN其实就是把前几名主播的id还有金币数量拼接成一个字符串--(area,topN)
        JavaRDD top3Rdd=groupRdd.map(new Function<Tuple2<String, Iterable<Tuple2<String, Integer>>>,Tuple2<String,String>>() {
            @Override
            public Tuple2<String, String> call(Tuple2<String, Iterable<Tuple2<String, Integer>>> tup) throws Exception {
                String area=tup._1;
                ArrayList<Tuple2<String, Integer>> tuple2s = Lists.newArrayList(tup._2);
                Collections.sort(tuple2s, new Comparator<Tuple2<String, Integer>>() {
                    @Override
                    public int compare(Tuple2<String, Integer> o1, Tuple2<String, Integer> o2) {
                        return o2._2 - o1._2;
                    }
                });

                StringBuffer buffer=new StringBuffer();
                for (int i = 0; i < tuple2s.size(); i++) {
                    if (i<3){
                        if (i != 0) {
                            buffer.append(",");
                        }
                        buffer.append(tuple2s.get(i)._1);
                        buffer.append(":");
                        buffer.append(tuple2s.get(i)._2);
                    }
                }
                return new Tuple2<String,String>(area,buffer.toString());
            }
        });
        //8、使用foreach将结果打印到控制台，多个字段使用制表符分割
        top3Rdd.foreach(new VoidFunction<Tuple2<String,String>>() {

            @Override
            public void call(Tuple2<String, String> tup) throws Exception {
                System.out.println(tup._1+"\t"+tup._2);
            }
        });
    }
}
